package org.fun.kafkademo.controller;

import org.fun.kafkademo.model.BatchProcessResult;
import org.fun.kafkademo.model.UserEvent;
import org.fun.kafkademo.producer.BatchProducer;
import org.fun.kafkademo.consumer.BatchConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

/**
 * Kafka批量处理控制器
 */
@RestController
@RequestMapping("/api/kafka")
public class KafkaBatchController {
    
    private static final Logger logger = LoggerFactory.getLogger(KafkaBatchController.class);
    
    @Autowired
    private BatchProducer batchProducer;
    
    @Autowired
    private BatchConsumer batchConsumer;
    
    /**
     * 生成并发送指定数量的用户事件
     */
    @PostMapping("/send-batch")
    public ResponseEntity<Map<String, Object>> sendBatchEvents(@RequestParam(defaultValue = "1000") int count) {
        logger.info("收到批量发送请求，数量: {}", count);
        
        try {
            // 生成用户事件数据
            List<UserEvent> events = batchProducer.generateUserEvents(count);
            
            // 批量发送
            BatchProcessResult result = batchProducer.sendBatchEvents(events);
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", true);
            response.put("message", "批量发送完成");
            response.put("result", result);
            
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            logger.error("批量发送失败", e);
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", false);
            response.put("message", "批量发送失败: " + e.getMessage());
            
            return ResponseEntity.internalServerError().body(response);
        }
    }
    
    /**
     * 异步生成并发送指定数量的用户事件
     */
    @PostMapping("/send-batch-async")
    public ResponseEntity<Map<String, Object>> sendBatchEventsAsync(@RequestParam(defaultValue = "1000") int count) {
        logger.info("收到异步批量发送请求，数量: {}", count);
        
        try {
            // 生成用户事件数据
            List<UserEvent> events = batchProducer.generateUserEvents(count);
            
            // 异步批量发送
            CompletableFuture<BatchProcessResult> future = CompletableFuture.supplyAsync(() -> {
                return batchProducer.sendBatchEvents(events);
            });
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", true);
            response.put("message", "异步批量发送已启动");
            response.put("count", count);
            
            // 异步处理结果
            future.whenComplete((result, throwable) -> {
                if (throwable != null) {
                    logger.error("异步批量发送失败", throwable);
                } else {
                    logger.info("异步批量发送完成: {}", result);
                }
            });
            
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            logger.error("启动异步批量发送失败", e);
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", false);
            response.put("message", "启动异步批量发送失败: " + e.getMessage());
            
            return ResponseEntity.internalServerError().body(response);
        }
    }
    
    /**
     * 发送单个用户事件
     */
    @PostMapping("/send-single")
    public ResponseEntity<Map<String, Object>> sendSingleEvent(@RequestBody UserEvent event) {
        logger.info("收到单个事件发送请求: {}", event);
        
        try {
            batchProducer.sendSingleEvent(event);
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", true);
            response.put("message", "单个事件发送成功");
            response.put("event", event);
            
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            logger.error("发送单个事件失败", e);
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", false);
            response.put("message", "发送单个事件失败: " + e.getMessage());
            
            return ResponseEntity.internalServerError().body(response);
        }
    }
    
    /**
     * 获取消费者处理统计信息
     */
    @GetMapping("/stats")
    public ResponseEntity<Map<String, Object>> getProcessingStats() {
        try {
            BatchProcessResult stats = batchConsumer.getProcessingStats();
            Map<String, Long> eventTypeCounts = batchConsumer.getEventTypeCounts();
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", true);
            response.put("processingStats", stats);
            response.put("eventTypeCounts", eventTypeCounts);
            
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            logger.error("获取统计信息失败", e);
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", false);
            response.put("message", "获取统计信息失败: " + e.getMessage());
            
            return ResponseEntity.internalServerError().body(response);
        }
    }
    
    /**
     * 重置消费者统计信息
     */
    @PostMapping("/reset-stats")
    public ResponseEntity<Map<String, Object>> resetStats() {
        try {
            batchConsumer.resetStats();
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", true);
            response.put("message", "统计信息已重置");
            
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            logger.error("重置统计信息失败", e);
            
            Map<String, Object> response = new HashMap<>();
            response.put("success", false);
            response.put("message", "重置统计信息失败: " + e.getMessage());
            
            return ResponseEntity.internalServerError().body(response);
        }
    }
    
    /**
     * 健康检查
     */
    @GetMapping("/health")
    public ResponseEntity<Map<String, Object>> health() {
        Map<String, Object> response = new HashMap<>();
        response.put("status", "UP");
        response.put("service", "Kafka Batch Demo");
        response.put("timestamp", System.currentTimeMillis());
        
        return ResponseEntity.ok(response);
    }
    
    /**
     * 获取API文档信息
     */
    @GetMapping("/info")
    public ResponseEntity<Map<String, Object>> getApiInfo() {
        Map<String, Object> response = new HashMap<>();
        response.put("service", "Kafka Batch Processing Demo");
        response.put("version", "1.0.0");
        Map<String, String> endpoints = new HashMap<>();
        endpoints.put("POST /api/kafka/send-batch", "批量发送用户事件 (参数: count)");
        endpoints.put("POST /api/kafka/send-batch-async", "异步批量发送用户事件 (参数: count)");
        endpoints.put("POST /api/kafka/send-single", "发送单个用户事件 (Body: UserEvent)");
        endpoints.put("GET /api/kafka/stats", "获取处理统计信息");
        endpoints.put("POST /api/kafka/reset-stats", "重置统计信息");
        endpoints.put("GET /api/kafka/health", "健康检查");
        endpoints.put("GET /api/kafka/info", "API信息");
        response.put("endpoints", endpoints);
        
        return ResponseEntity.ok(response);
    }
}
